package org.ala.tiktools.rpc.netty.handler;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.msg.RpcMsg;
import org.ala.tiktools.rpc.msg.login.RpcLoginBody;
import org.ala.tiktools.rpc.msg.login.RpcLoginErrorMsg;
import org.ala.tiktools.rpc.msg.login.RpcLoginMsg;
import org.ala.tiktools.rpc.msg.login.RpcLoginNoMsg;
import org.ala.tiktools.rpc.msg.login.RpcLoginSuccessMsg;
import org.ala.tiktools.rpc.msg.push.RpcPushMsg;
import org.ala.tiktools.rpc.plugin.serializer.SerializeThreadPool;
import org.ala.tiktools.rpc.plugin.serializer.kryo.KryoThreadSerializer;
import org.slf4j.Logger;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 登录消息拦截器
 * <p>  只拦截push消息
 *
 * @author ala
 * @date 2025-04-09 10:18
 */
public class RpcLoginInterceptor extends ChannelInboundHandlerAdapter {


    static Logger log = LoggerFactory.getLogger("rpc", "handler");


    /**
     *  对象创建时间 == channel初始化时间 约等于（只多不少） 三次握手成功时间戳
     */
    protected long connectTimestamp;
    /**
     *  登录是否成功
     */
    protected boolean login = false;


    public RpcLoginInterceptor() {
        connectTimestamp = System.currentTimeMillis();
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //  规定时间后，检测登录状态
        ctx.channel().eventLoop().schedule(() -> {
            if (!login) {
                SocketAddress addr = ctx.channel().remoteAddress();
                log.info("[RpcLoginInterceptor channelActive] 远程主机登录超时，关闭连接 remoteAddress:{}", addr.toString());
                try {
                    ctx.channel().close().sync();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, RpcConfigInfo.INSTANCE.getAuth().getTimeout(), TimeUnit.MILLISECONDS);

        super.channelActive(ctx);
    }


    @SuppressWarnings("rawtypes")
	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcMsg m = (RpcMsg) msg;

        //  如果是Login消息
        if (m instanceof RpcLoginMsg) {
        	checkAndHandlyLogin(ctx, (RpcLoginMsg) m);
        }
        //  如果是Push消息
        else if (m instanceof RpcPushMsg) {
            if (login) {
                super.channelRead(ctx, msg);
            } else {
                //  大概率收不到这种消息，除非还没登录，在timeout期内，就开始疯狂发消息
            	//	udp会有这种情况，而且udp下对面还感知不到这边已经关channel了
                log.info("[RpcLoginInterceptor channelActive] 远程主机还未登录，localAddress:{} remoteAddress:{} channelId:{} msg:{}", ctx.channel().localAddress(), ctx.channel().remoteAddress(), ctx.channel().id(), m.getHead());
                RpcLoginNoMsg ack = RpcLoginNoMsg.build();
                ack.getHead().id(m.getHead().getId());
                ctx.channel().writeAndFlush(ack);
            }
        }
        //  其他消息不予拦截
        else {
            super.channelRead(ctx, msg);
        }
    }


    /**
     *  处理登录消息
     */
    protected void checkAndHandlyLogin(ChannelHandlerContext ctx, RpcLoginMsg msg) {
    	//	在eventLoop线程反序列化
    	if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
        	RpcLoginBody body = KryoThreadSerializer.INSTANCE.deSerialize(msg.getBodyBytes(), RpcLoginBody.class);
        	//  检测登录是否成功
        	String username = body.getUsername(), password = body.getPassword();
        	//	登录成功
        	if (RpcConfigInfo.INSTANCE.getAuth().getUsername().equals(username)
        			&& RpcConfigInfo.INSTANCE.getAuth().getPassword().equals(password)) {
        		this.login = true;
        		log.info("[RpcLoginInterceptor channelActive] 远程主机登录成功，localAddress:{} remoteAddress:{} channelId:{}", ctx.channel().localAddress(), ctx.channel().remoteAddress(), ctx.channel().id());
        		RpcLoginSuccessMsg ack = RpcLoginSuccessMsg.build();
        		ack.getHead().id(msg.getHead().getId());
        		ctx.channel().writeAndFlush(ack);
        		//	不再往右继续触发
//              super.channelRead(ctx, msg);
        	} 
        	//	登录失败
        	else {
        		RpcLoginErrorMsg ack = RpcLoginErrorMsg.build();
        		ack.getHead().id(msg.getHead().getId());
        		ctx.channel().writeAndFlush(ack);
        		ctx.channel().close();
        		log.info("[RpcLoginInterceptor channelActive] 远程主机登录失败，关闭连接  localAddress:{} remoteAddress:{} channelId:{}", ctx.channel().localAddress(), ctx.channel().remoteAddress(), ctx.channel().id());
        	}
    	}
    	//	在序列化线程中执行反序列化
    	else {
        	SerializeThreadPool.INSTANCE.deSerlize(msg.getBodyBytes(), RpcLoginBody.class, body -> {
		          //  检测登录是否成功
		          String username = body.getUsername(), password = body.getPassword();
		          //	登录成功
		          if (RpcConfigInfo.INSTANCE.getAuth().getUsername().equals(username)
		              && RpcConfigInfo.INSTANCE.getAuth().getPassword().equals(password)) {
		              this.login = true;
		          	log.info("[RpcLoginInterceptor channelActive] 远程主机登录成功，localAddress:{} remoteAddress:{} channelId:{}", ctx.channel().localAddress(), ctx.channel().remoteAddress(), ctx.channel().id());
		              RpcLoginSuccessMsg ack = RpcLoginSuccessMsg.build();
		              ack.getHead().id(msg.getHead().getId());
		              ctx.channel().writeAndFlush(ack);
		              //	不再往右继续触发
		//              super.channelRead(ctx, msg);
		          } 
		          //	登录失败
		          else {
		          	 RpcLoginErrorMsg ack = RpcLoginErrorMsg.build();
		               ack.getHead().id(msg.getHead().getId());
		               ctx.channel().writeAndFlush(ack);
		               ctx.channel().close();
		               log.info("[RpcLoginInterceptor channelActive] 远程主机登录失败，关闭连接  localAddress:{} remoteAddress:{} channelId:{}", ctx.channel().localAddress(), ctx.channel().remoteAddress(), ctx.channel().id());
		          }
		  	});
    	}
    }

}
